package cn.itcast.model.matchtag

import java.util.Properties

import cn.itcast.model.bean.{HBaseMeta, TagRule}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession, functions}

object GenderModel {
  def main(args: Array[String]): Unit = {
    //1. 获取SparkSession对象
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("GenderModel")
      .getOrCreate()
    import spark.implicits._
    val url:String="jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=" +
      "UTF-8&serverTimezone=UTC&user=root&password=123456"
    val table = "tbl_basic_tag"
    val properties=new Properties()
    val mysqlDF: DataFrame = spark.read.jdbc(url,table,properties)
    mysqlDF.show()
    val fourDF: Dataset[Row] = mysqlDF.select('id,'rule).where("id=385")
    fourDF.show()
    val meta = fourDF.map(row => {
      val rule: String = row.getAs[String]("rule")
      val tmp: Array[String] = rule.split("##")
      val tuples: Array[(String, String)] = tmp.map(kv => {
        val arr: Array[String] = kv.split("=")
        (arr(0), arr(1))
      })
      val map = tuples.toMap
      HBaseMeta(map)
    }).collect()(0)
    println(meta)
    //   3. 加载**性别5级规则**数据(id,rule) (男/女)
    //      主要是为了后面给用户打标签使用
    val fiveDS: Dataset[Row] = mysqlDF.select('id,'rule).where("pid=385")
    val fiveList: List[TagRule] = fiveDS.map(row => {
      val id = row.getAs[Int]("id").toString
      val rule = row.getAs[String]("rule").toString
      TagRule(id, rule)
    }).collect().toList
    println(fiveList)
    // 如何从HBase中加载数据?
    // 1. 利用HBase原生的API进行获取.
    // 2. HBase整合Phoenix,JDBC
    // 3. 将HBase作为一个数据源,利用spark.read()直接加载数据.
    //   使用**自定义数据**源进行数据的加载
    val hbaseSource: DataFrame = spark.read
      //      .format("数据源的全路径名")
      .format("cn.itcast.up.model.HBaseSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, meta.hbaseTable)
      .option(HBaseMeta.FAMILY, meta.family)
      .option(HBaseMeta.SELECTFIELDS, meta.selectFields)
      .load()
   // hbaseSource.show()
    //4. 开始进行标签计算
    //   编写一个自定义函数,在函数中实现数据的匹配操作.
    // 编写UDF自定义函数的时候, 函数一定要有返回值.
    val matchTag:UserDefinedFunction=functions.udf((gender:String)=>{
      var tagID=""
      for(tag<-fiveList){
        if(tag.rule.equals(gender)) tagID=tag.id
      }
      tagID
    })
    val result: DataFrame = hbaseSource.select('id.as("userId"), matchTag('gender).as("tagIds"))
    result.show()

    //5. 将数据落地,将结果保存到HBase
    //   使用**自定义数据**源进行数据的落地操作
    result.write
      //      .format("自定义数据落地的数据源")
      .format("cn.itcast.model.utils.HBaseSource")
      //      .option("参数名", "参数值")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test_bs")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagIds")
      .save()
  }
}
